# Copyright 2024 Marimo. All rights reserved.
from __future__ import annotations

from typing import Any
from unittest.mock import patch

from marimo._ast.cell import RuntimeStateType
from marimo._data.models import Database, DataTable, DataTableColumn, Schema
from marimo._messaging.cell_output import CellChannel, CellOutput
from marimo._messaging.msgspec_encoder import asdict as serialize
from marimo._messaging.ops import (
    CellOp,
    Datasets,
    DataSourceConnection,
    DataSourceConnections,
    InstallingPackageAlert,
    SendUIElementMessage,
    SQLMetadata,
    SQLTableListPreview,
    SQLTablePreview,
    StartupLogs,
    UpdateCellCodes,
    UpdateCellIdsRequest,
    VariableDeclaration,
    Variables,
    VariableValue,
    VariableValues,
    serialize_kernel_message,
)
from marimo._runtime.requests import (
    CreationRequest,
    ExecuteMultipleRequest,
    ExecutionRequest,
    SetUIElementValueRequest,
)
from marimo._server.session.session_view import SessionView
from marimo._sql.engines.duckdb import INTERNAL_DUCKDB_ENGINE
from marimo._types.ids import CellId_t, RequestId, VariableName, WidgetModelId
from marimo._utils.parse_dataclass import parse_raw

cell_id = CellId_t("cell_1")

initial_output = CellOutput(
    channel=CellChannel.OUTPUT,
    data="Initial output",
    mimetype="text/plain",
)
updated_output = CellOutput(
    channel=CellChannel.OUTPUT,
    data="Updated output",
    mimetype="text/plain",
)

initial_status: RuntimeStateType = "running"
updated_status: RuntimeStateType = "running"


def test_cell_ids(session_view: SessionView) -> None:
    assert session_view.cell_ids is None

    session_view.add_operation(
        UpdateCellIdsRequest(
            cell_ids=[cell_id],
        )
    )
    operation = session_view.operations[0]
    assert isinstance(operation, UpdateCellIdsRequest)
    assert operation.cell_ids == [cell_id]


def test_session_view_cell_op(session_view: SessionView) -> None:
    # Create initial CellOp
    initial_cell_op = CellOp(
        cell_id=cell_id, output=initial_output, status=initial_status
    )
    session_view.add_operation(initial_cell_op)

    # Add updated CellOp to SessionView
    updated_cell_op = CellOp(
        cell_id=cell_id, output=updated_output, status=updated_status
    )
    session_view.add_operation(updated_cell_op)

    assert session_view.cell_operations[cell_id].output == updated_output
    assert session_view.cell_operations[cell_id].status == updated_status


# Test adding Variables to SessionView
def test_session_view_variables(session_view: SessionView) -> None:
    # Create Variables operation
    variables_op = Variables(
        variables=[
            VariableDeclaration(name="var1", declared_by=[], used_by=[])
        ]
    )
    session_view.add_operation(variables_op)

    # Check if the Variables operation was added correctly
    assert session_view.variable_operations == variables_op


# Test adding VariableValues to SessionView
def test_session_view_variable_values(session_view: SessionView) -> None:
    # Create Variables operation
    variables_op = Variables(
        variables=[
            VariableDeclaration(
                name="var1", declared_by=[cell_id], used_by=[]
            ),
            VariableDeclaration(
                name="var2", declared_by=[cell_id], used_by=[]
            ),
        ]
    )
    session_view.add_operation(variables_op)

    # Create VariableValues operation
    variable_values_op = VariableValues(
        variables=[
            VariableValue.create(name="var1", value=1),
            VariableValue.create(name="var2", value="hello"),
        ]
    )
    session_view.add_operation(variable_values_op)

    variables_names = session_view.variable_values.keys()
    assert list(variables_names) == ["var1", "var2"]

    # Add new Variable operation without the previous variables
    variables_op = Variables(
        variables=[
            VariableDeclaration(
                name="var2", declared_by=[cell_id], used_by=[]
            ),
            VariableDeclaration(
                name="var3", declared_by=[cell_id], used_by=[]
            ),
        ]
    )
    session_view.add_operation(variables_op)

    variables_names = session_view.variable_values.keys()
    # var1 was removed, var2 was not changed, var3 has no value yet
    assert list(variables_names) == ["var2"]


def test_ui_values(session_view: SessionView) -> None:
    session_view.add_control_request(
        SetUIElementValueRequest.from_ids_and_values([("test_ui", 123)])
    )
    assert "test_ui" in session_view.ui_values
    assert session_view.ui_values["test_ui"] == 123

    # Can add multiple values
    # and can overwrite values
    session_view.add_control_request(
        SetUIElementValueRequest.from_ids_and_values(
            [("test_ui2", 456), ("test_ui", 789)]
        )
    )
    assert "test_ui2" in session_view.ui_values
    assert "test_ui" in session_view.ui_values
    assert session_view.ui_values["test_ui2"] == 456
    assert session_view.ui_values["test_ui"] == 789

    # Can add from CreationRequest
    session_view.add_control_request(
        CreationRequest(
            execution_requests=(),
            set_ui_element_value_request=SetUIElementValueRequest.from_ids_and_values(
                [("test_ui3", 101112)]
            ),
            auto_run=True,
        )
    )
    assert "test_ui3" in session_view.ui_values


def test_model_message_values(session_view: SessionView) -> None:
    model_id = WidgetModelId("test_model")
    model_id2 = WidgetModelId("test_model2")

    session_view.add_operation(
        SendUIElementMessage(
            model_id=model_id,
            message={"key": "value"},
            ui_element=None,
        )
    )
    assert model_id in session_view.model_messages
    assert session_view.model_messages[model_id][0].message == {"key": "value"}

    # Can add to existing model
    session_view.add_operation(
        SendUIElementMessage(
            model_id=model_id,
            message={"key": "new_value"},
            ui_element=None,
        )
    )
    assert len(session_view.model_messages[model_id]) == 2
    assert session_view.model_messages[model_id][1].message == {
        "key": "new_value"
    }

    # Can add multiple models
    session_view.add_operation(
        SendUIElementMessage(
            model_id=model_id2,
            message={"key2": "value2"},
            ui_element=None,
        )
    )
    assert model_id2 in session_view.model_messages
    assert session_view.model_messages[model_id2][0].message == {
        "key2": "value2"
    }


def test_last_run_code(session_view: SessionView) -> None:
    session_view.add_control_request(
        ExecuteMultipleRequest(
            cell_ids=[cell_id],
            codes=["print('hello')"],
        )
    )
    assert session_view.last_executed_code[cell_id] == "print('hello')"

    # Can overwrite values and add multiple
    session_view.add_control_request(
        ExecuteMultipleRequest(
            cell_ids=[cell_id, "cell_2"],
            codes=["print('hello world')", "print('hello world')"],
        )
    )
    assert session_view.last_executed_code[cell_id] == "print('hello world')"
    assert session_view.last_executed_code["cell_2"] == "print('hello world')"

    # Can add from CreationRequest
    session_view.add_control_request(
        CreationRequest(
            execution_requests=(
                ExecutionRequest(cell_id=cell_id, code="print('hello')"),
            ),
            set_ui_element_value_request=SetUIElementValueRequest.from_ids_and_values(
                []
            ),
            auto_run=True,
        )
    )
    assert session_view.last_executed_code[cell_id] == "print('hello')"


def test_serialize_parse_variable_value() -> None:
    original = VariableValue.create(name="var1", value=1)
    serialized = serialize(original)
    assert serialized == {"datatype": "int", "name": "var1", "value": "1"}
    parsed = parse_raw(serialized, VariableValue)
    assert parsed == original


def test_add_variables(session_view: SessionView) -> None:
    session_view.add_raw_operation(
        serialize_kernel_message(
            Variables(
                variables=[
                    VariableDeclaration(
                        name="var1", declared_by=[cell_id], used_by=[]
                    ),
                    VariableDeclaration(
                        name="var2", declared_by=[cell_id], used_by=[]
                    ),
                ]
            )
        )
    )
    session_view.add_raw_operation(
        serialize_kernel_message(
            VariableValues(
                variables=[
                    VariableValue.create(name="var1", value=1),
                    VariableValue.create(name="var2", value="hello"),
                ]
            )
        )
    )

    assert session_view.variable_operations.variables[0].name == "var1"
    assert session_view.variable_operations.variables[1].name == "var2"
    assert session_view.variable_values["var1"].value == "1"
    assert session_view.variable_values["var1"].datatype == "int"
    assert session_view.variable_values["var2"].value == "hello"
    assert session_view.variable_values["var2"].datatype == "str"


def test_add_datasets(session_view: SessionView) -> None:
    session_view.add_raw_operation(
        serialize_kernel_message(
            Datasets(
                tables=[
                    DataTable(
                        source_type="local",
                        source="df",
                        name="table1",
                        columns=[
                            DataTableColumn(
                                name="col1",
                                type="boolean",
                                external_type="BOOL",
                                sample_values=["true", "false"],
                            )
                        ],
                        num_rows=1,
                        num_columns=1,
                        variable_name="df1",
                    ),
                    DataTable(
                        source_type="local",
                        source="df",
                        name="table2",
                        columns=[
                            DataTableColumn(
                                name="col2",
                                type="integer",
                                external_type="INT",
                                sample_values=["1", "2"],
                            )
                        ],
                        num_rows=2,
                        num_columns=2,
                        variable_name="df2",
                    ),
                ]
            )
        )
    )

    assert session_view.datasets.tables[0].name == "table1"
    assert session_view.datasets.tables[1].name == "table2"
    assert session_view.datasets.tables[0].variable_name == "df1"
    assert session_view.datasets.tables[1].variable_name == "df2"

    # Can add a new table and overwrite an existing table

    session_view.add_raw_operation(
        serialize_kernel_message(
            Datasets(
                tables=[
                    DataTable(
                        source_type="local",
                        source="df",
                        name="table2",
                        columns=[
                            DataTableColumn(
                                name="new_col",
                                type="boolean",
                                external_type="BOOL",
                                sample_values=["true", "false"],
                            )
                        ],
                        num_rows=20,
                        num_columns=20,
                        variable_name="df2",
                    ),
                    DataTable(
                        source_type="local",
                        source="df",
                        name="table3",
                        columns=[],
                        num_rows=3,
                        num_columns=3,
                        variable_name="df3",
                    ),
                ]
            )
        )
    )

    assert session_view.datasets.tables[0].name == "table1"
    # Updated
    assert session_view.datasets.tables[1].name == "table2"
    assert session_view.datasets.tables[1].columns[0].name == "new_col"
    assert session_view.datasets.tables[1].num_rows == 20
    # Added
    assert session_view.datasets.tables[2].name == "table3"
    assert session_view.datasets.tables[2].variable_name == "df3"

    # Can filter out tables from new variables
    session_view.add_raw_operation(
        serialize_kernel_message(
            Variables(
                variables=[
                    VariableDeclaration(
                        name="df2", declared_by=[cell_id], used_by=[]
                    ),
                ]
            )
        )
    )

    assert len(session_view.datasets.tables) == 1
    assert session_view.datasets.tables[0].name == "table2"


def test_add_datasets_clear_channel(session_view: SessionView) -> None:
    session_view.add_raw_operation(
        serialize_kernel_message(
            Datasets(
                tables=[
                    DataTable(
                        source_type="duckdb",
                        source="db",
                        name="db.table1",
                        columns=[],
                        num_rows=1,
                        num_columns=1,
                        variable_name=None,
                    ),
                    DataTable(
                        source_type="local",
                        source="memory",
                        name="df1",
                        columns=[],
                        num_rows=1,
                        num_columns=1,
                        variable_name="df1",
                    ),
                ],
                clear_channel="duckdb",
            )
        )
    )

    assert len(session_view.datasets.tables) == 2
    names = [t.name for t in session_view.datasets.tables]
    assert "db.table1" in names
    assert "df1" in names

    session_view.add_raw_operation(
        serialize_kernel_message(
            Datasets(
                tables=[
                    DataTable(
                        source_type="local",
                        source="db",
                        name="db.table2",
                        columns=[],
                        num_rows=1,
                        num_columns=1,
                        variable_name=None,
                    )
                ],
                clear_channel="duckdb",
            )
        )
    )

    assert len(session_view.datasets.tables) == 2
    names = [t.name for t in session_view.datasets.tables]
    assert "db.table1" not in names
    assert "df1" in names
    assert "db.table2" in names


def test_add_data_source_connections(session_view: SessionView) -> None:
    # Add initial connections
    session_view.add_raw_operation(
        serialize_kernel_message(
            DataSourceConnections(
                connections=[
                    DataSourceConnection(
                        source="duckdb",
                        dialect="duckdb",
                        name="db1",
                        display_name="duckdb (db1)",
                        databases=[],
                    ),
                    DataSourceConnection(
                        source="sqlalchemy",
                        dialect="postgresql",
                        name="pg1",
                        display_name="postgresql (pg1)",
                        databases=[],
                    ),
                    DataSourceConnection(
                        source="duckdb",
                        dialect="default",
                        name=INTERNAL_DUCKDB_ENGINE,
                        display_name="duckdb internal",
                        databases=[],
                    ),
                ]
            )
        )
    )

    assert len(session_view.data_connectors.connections) == 3
    names = [c.name for c in session_view.data_connectors.connections]
    assert "db1" in names
    assert "pg1" in names
    assert INTERNAL_DUCKDB_ENGINE in names

    # Add new connection and update existing
    session_view.add_raw_operation(
        serialize_kernel_message(
            DataSourceConnections(
                connections=[
                    DataSourceConnection(
                        source="duckdb",
                        dialect="duckdb",
                        name="db1",
                        display_name="duckdb (db1_updated)",
                        databases=[],
                    ),
                    DataSourceConnection(
                        source="sqlalchemy",
                        dialect="mysql",
                        name="mysql1",
                        display_name="mysql (mysql1)",
                        databases=[],
                    ),
                ]
            )
        )
    )

    assert len(session_view.data_connectors.connections) == 4
    conns = {c.name: c for c in session_view.data_connectors.connections}

    # Check updated connection
    assert "db1" in conns
    assert conns["db1"].display_name == "duckdb (db1_updated)"

    # Check new connection replaced old one
    assert "mysql1" in conns
    assert conns["mysql1"].dialect == "mysql"
    # Check existing connection
    assert "pg1" in conns
    assert INTERNAL_DUCKDB_ENGINE in names

    # Check connectors in operations
    assert session_view.data_connectors in session_view.operations

    # Filter out connections from variables
    session_view.add_raw_operation(
        serialize_kernel_message(
            Variables(
                variables=[
                    VariableDeclaration(
                        name="mysql1", declared_by=[cell_id], used_by=[]
                    ),
                ]
            )
        )
    )
    assert len(session_view.data_connectors.connections) == 2
    session_view_names = [
        c.name for c in session_view.data_connectors.connections
    ]
    assert "mysql1" in session_view_names
    assert INTERNAL_DUCKDB_ENGINE in session_view_names


def test_add_sql_table_previews() -> None:
    session_view = SessionView()

    # Add initial connections
    session_view.add_raw_operation(
        serialize_kernel_message(
            DataSourceConnections(
                connections=[
                    DataSourceConnection(
                        source="duckdb",
                        name="connection1",
                        dialect="duckdb",
                        display_name="duckdb (connection1)",
                        databases=[
                            Database(
                                name="db1",
                                dialect="duckdb",
                                schemas=[
                                    Schema(
                                        name="db1",
                                        tables=[
                                            DataTable(
                                                name="table1",
                                                source_type="connection",
                                                source="db1",
                                                columns=[],
                                                num_rows=0,
                                                num_columns=0,
                                                variable_name=None,
                                            )
                                        ],
                                    )
                                ],
                            )
                        ],
                    )
                ],
            )
        )
    )

    session_view_connections = session_view.data_connectors.connections
    assert session_view_connections[0].databases[0].schemas[0].tables == [
        DataTable(
            source_type="connection",
            source="db1",
            name="table1",
            num_rows=0,
            num_columns=0,
            variable_name=None,
            columns=[],
        )
    ]

    session_view.add_raw_operation(
        serialize_kernel_message(
            SQLTablePreview(
                metadata=SQLMetadata(
                    connection="connection1", database="db1", schema="db1"
                ),
                request_id=RequestId("request_id"),
                table=DataTable(
                    name="table1",
                    source_type="connection",
                    source="db1",
                    num_rows=10,  # Updated
                    num_columns=0,
                    variable_name=None,
                    columns=[],
                ),
            )
        )
    )
    session_view_connections = session_view.data_connectors.connections
    assert (
        session_view_connections[0].databases[0].schemas[0].tables[0].num_rows
        == 10
    )

    # Add sql table preview list
    session_view.add_raw_operation(
        serialize_kernel_message(
            SQLTableListPreview(
                metadata=SQLMetadata(
                    connection="connection1", database="db1", schema="db1"
                ),
                request_id=RequestId("request_id"),
                tables=[
                    DataTable(
                        name="table2",
                        source_type="connection",
                        source="db1",
                        num_rows=20,
                        num_columns=10,
                        variable_name=VariableName("var"),
                        columns=[],
                    )
                ],
            )
        )
    )

    assert session_view_connections[0].databases[0].schemas[0].tables == [
        DataTable(
            source_type="connection",
            source="db1",
            name="table2",
            num_rows=20,
            num_columns=10,
            variable_name=VariableName("var"),
            columns=[],
            engine=None,
            type="table",
            primary_keys=None,
            indexes=None,
        )
    ]


def test_add_cell_op(session_view: SessionView) -> None:
    session_view.add_raw_operation(
        serialize_kernel_message(
            CellOp(
                cell_id=cell_id, output=initial_output, status=initial_status
            )
        )
    )

    assert session_view.cell_operations[cell_id].output == initial_output
    assert session_view.cell_operations[cell_id].status == initial_status


# patch time
@patch("time.time", return_value=123)
def test_combine_console_outputs(
    time_mock: Any, session_view: SessionView
) -> None:
    del time_mock
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("one"),
            status="running",
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("two"),
            status="running",
        )
    )

    # Consecutive text/plain stdout outputs are merged
    assert session_view.cell_operations[cell_id].console == [
        CellOutput.stdout("onetwo"),
    ]

    # Moves to queued
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=None,
            status="queued",
        )
    )

    assert session_view.cell_operations[cell_id].console == [
        CellOutput.stdout("onetwo"),
    ]

    # Moves to running clears console
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=None,
            status="running",
        )
    )
    assert session_view.cell_operations[cell_id].console == []

    # Write again
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("three"),
            status="running",
        )
    )
    assert session_view.cell_operations[cell_id].console == [
        CellOutput.stdout("three")
    ]


@patch("time.time", return_value=123)
def test_stdin(time_mock: Any, session_view: SessionView) -> None:
    del time_mock
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("Hello"),
            status="running",
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdin("What is your name?"),
            status="running",
        )
    )

    assert session_view.cell_operations[cell_id].console == [
        CellOutput.stdout("Hello"),
        CellOutput.stdin("What is your name?"),
    ]

    session_view.add_stdin("marimo")

    assert session_view.cell_operations[cell_id].console == [
        CellOutput.stdout("Hello"),
        CellOutput.stdout("What is your name? marimo\n"),
    ]


@patch("time.time", return_value=123)
def test_merge_consecutive_text_plain_outputs(
    time_mock: Any, session_view: SessionView
) -> None:
    """Test that consecutive text/plain outputs with same channel are merged."""
    del time_mock

    # Add multiple consecutive stdout outputs
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("Hello "),
            status="running",
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("World"),
            status="running",
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("!"),
            status="running",
        )
    )

    # Should be merged into a single output
    assert len(session_view.cell_operations[cell_id].console) == 1
    assert (
        session_view.cell_operations[cell_id].console[0].data == "Hello World!"
    )
    assert (
        session_view.cell_operations[cell_id].console[0].channel
        == CellChannel.STDOUT
    )


@patch("time.time", return_value=123)
def test_merge_different_channels_not_merged(
    time_mock: Any, session_view: SessionView
) -> None:
    """Test that outputs with different channels are not merged."""
    del time_mock

    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("stdout message"),
            status="running",
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stderr("stderr message"),
            status="running",
        )
    )

    # Should remain separate
    assert len(session_view.cell_operations[cell_id].console) == 2
    assert (
        session_view.cell_operations[cell_id].console[0].channel
        == CellChannel.STDOUT
    )
    assert (
        session_view.cell_operations[cell_id].console[1].channel
        == CellChannel.STDERR
    )


@patch("time.time", return_value=123)
def test_merge_different_mimetypes_not_merged(
    time_mock: Any, session_view: SessionView
) -> None:
    """Test that outputs with different mimetypes are not merged."""
    del time_mock

    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("plain text", mimetype="text/plain"),
            status="running",
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("html content", mimetype="text/html"),
            status="running",
        )
    )

    # Should remain separate
    assert len(session_view.cell_operations[cell_id].console) == 2
    assert (
        session_view.cell_operations[cell_id].console[0].mimetype
        == "text/plain"
    )
    assert (
        session_view.cell_operations[cell_id].console[1].mimetype
        == "text/html"
    )


@patch("time.time", return_value=123)
def test_merge_with_non_string_data_not_merged(
    time_mock: Any, session_view: SessionView
) -> None:
    """Test that outputs with non-string data are not merged."""
    del time_mock

    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput.stdout("text"),
            status="running",
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=CellOutput(
                channel=CellChannel.STDOUT,
                mimetype="text/plain",
                data={"key": "value"},  # dict, not string
            ),
            status="running",
        )
    )

    # Should remain separate
    assert len(session_view.cell_operations[cell_id].console) == 2


@patch("time.time", return_value=123)
def test_get_cell_outputs(time_mock: Any, session_view: SessionView) -> None:
    del time_mock
    cell_2_id = "cell_2"
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            output=initial_output,
            status=initial_status,
        ),
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_2_id,
            output=None,
            status=updated_status,
        ),
    )

    assert session_view.get_cell_outputs([cell_id]) == {
        cell_id: initial_output
    }
    assert session_view.get_cell_outputs([cell_id, cell_2_id]) == {
        cell_id: initial_output
    }

    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            output=updated_output,
            status=updated_status,
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_2_id,
            output=updated_output,
            status=updated_status,
        )
    )

    assert session_view.get_cell_outputs([cell_id, cell_2_id]) == {
        cell_id: updated_output,
        cell_2_id: updated_output,
    }


@patch("time.time", return_value=123)
def test_get_cell_console_outputs(
    time_mock: Any, session_view: SessionView
) -> None:
    del time_mock
    cell_2_id = "cell_2"
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=[CellOutput.stdout("one")],
            status=initial_status,
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_2_id,
            console=None,
            status=updated_status,
        )
    )

    assert session_view.get_cell_console_outputs([cell_id]) == {
        cell_id: [CellOutput.stdout("one")]
    }
    assert session_view.get_cell_console_outputs([cell_id, cell_2_id]) == {
        cell_id: [CellOutput.stdout("one")],
    }

    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            console=[CellOutput.stdout("two")],
            status=updated_status,
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id=cell_2_id,
            console=[CellOutput.stdout("two")],
            status=updated_status,
        )
    )

    # Consecutive text/plain stdout outputs are merged
    assert session_view.get_cell_console_outputs([cell_id, cell_2_id]) == {
        cell_id: [CellOutput.stdout("onetwo")],
        cell_2_id: [CellOutput.stdout("two")],
    }


def test_mark_auto_export(session_view: SessionView):
    assert session_view.needs_export("html")
    assert session_view.needs_export("md")

    session_view.mark_auto_export_html()
    assert not session_view.needs_export("html")

    session_view.mark_auto_export_md()
    assert not session_view.needs_export("md")

    session_view._touch()
    assert session_view.needs_export("html")
    assert session_view.needs_export("md")

    session_view.mark_auto_export_html()
    session_view.mark_auto_export_md()
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            output=initial_output,
            status=initial_status,
        ),
    )
    assert session_view.needs_export("html")
    assert session_view.needs_export("md")

    session_view.mark_auto_export_session()
    assert not session_view.needs_export("session")

    session_view._touch()
    assert session_view.needs_export("session")


def test_stale_code(session_view: SessionView) -> None:
    """Test that stale code is properly tracked and included in operations."""
    assert session_view.stale_code is None

    # Add stale code operation
    stale_code_op = UpdateCellCodes(
        cell_ids=["cell1"],
        codes=["print('hello')"],
        code_is_stale=True,
    )
    session_view.add_operation(stale_code_op)

    # Verify stale code is tracked
    assert session_view.stale_code == stale_code_op
    assert session_view.stale_code in session_view.operations

    # Add non-stale code operation
    non_stale_code_op = UpdateCellCodes(
        cell_ids=["cell2"],
        codes=["print('world')"],
        code_is_stale=False,
    )
    session_view.add_operation(non_stale_code_op)

    # Verify non-stale code doesn't affect stale_code tracking
    assert session_view.stale_code == stale_code_op
    assert session_view.stale_code in session_view.operations

    # Update stale code
    new_stale_code_op = UpdateCellCodes(
        cell_ids=["cell3"],
        codes=["print('updated')"],
        code_is_stale=True,
    )
    session_view.add_operation(new_stale_code_op)

    # Verify stale code is updated
    assert session_view.stale_code == new_stale_code_op
    assert session_view.stale_code in session_view.operations
    assert stale_code_op not in session_view.operations


def test_dataset_filter_by_engine_and_variable(
    session_view: SessionView,
) -> None:
    # Initially add three tables: one with an engine, one with a variable name, and one with neither
    session_view.add_raw_operation(
        serialize_kernel_message(
            Datasets(
                tables=[
                    DataTable(
                        source_type="connection",
                        source="duckdb",
                        name="table_with_engine",
                        columns=[],
                        num_rows=1,
                        num_columns=1,
                        engine="some_engine",
                        variable_name=None,
                    ),
                    DataTable(
                        source_type="local",
                        source="df",
                        name="table_with_var",
                        columns=[],
                        num_rows=1,
                        num_columns=1,
                        engine=None,
                        variable_name="some_var",
                    ),
                    DataTable(
                        source_type="local",
                        source="df",
                        name="table_none",
                        columns=[],
                        num_rows=1,
                        num_columns=1,
                        engine=None,
                        variable_name=None,
                    ),
                ]
            )
        )
    )
    # Confirm all three are present before filtering
    assert len(session_view.datasets.tables) == 3

    # Step 1: Add operation of all variables
    session_view.add_operation(
        Variables(
            variables=[
                VariableDeclaration(
                    name="some_engine", declared_by=[], used_by=[]
                ),
                VariableDeclaration(
                    name="some_var", declared_by=[], used_by=[]
                ),
            ]
        )
    )
    assert len(session_view.datasets.tables) == 3

    # Step 2: Only "some_engine" is in scope => keep table_with_engine + table_none
    session_view.add_operation(
        Variables(
            variables=[
                VariableDeclaration(
                    name="some_engine", declared_by=[], used_by=[]
                )
            ]
        )
    )
    table_names = [t.name for t in session_view.datasets.tables]
    assert "table_with_engine" in table_names
    assert "table_with_var" not in table_names
    assert "table_none" in table_names

    # Step 3: No variables => only table with neither engine nor variable_name is kept
    session_view.add_operation(Variables(variables=[]))
    table_names = [t.name for t in session_view.datasets.tables]
    assert table_names == ["table_none"]


def test_is_empty(session_view: SessionView) -> None:
    """Test that SessionView.is_empty() correctly detects empty session views."""

    # Initially empty
    assert session_view.is_empty()

    # Add a cell operation without output or console
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            status=initial_status,
        )
    )

    # Add a cell operation
    session_view.add_operation(
        CellOp(
            cell_id=cell_id,
            output=initial_output,
            status=initial_status,
        )
    )

    # No longer empty
    assert not session_view.is_empty()


def test_is_empty_multiple_operations(session_view: SessionView) -> None:
    assert session_view.is_empty()

    # Add multiple operations - should still not be empty
    session_view.add_operation(
        CellOp(
            cell_id="cell1",
            output=initial_output,
            status="idle",
        )
    )
    session_view.add_operation(
        CellOp(
            cell_id="cell2",
            output=updated_output,
            status="idle",
        )
    )
    assert not session_view.is_empty()


def test_session_view_startup_logs(session_view: SessionView) -> None:
    # Test adding a startup log with "start" status
    start_log = StartupLogs(content="Starting process...", status="start")
    session_view.add_operation(start_log)

    assert session_view.startup_logs is not None
    assert session_view.startup_logs.content == "Starting process..."
    assert session_view.startup_logs.status == "start"

    # Test appending to startup log
    append_log = StartupLogs(content=" more content", status="append")
    session_view.add_operation(append_log)

    assert session_view.startup_logs is not None
    assert (
        session_view.startup_logs.content == "Starting process... more content"
    )
    assert session_view.startup_logs.status == "append"

    # Test marking startup log as done
    done_log = StartupLogs(content=" done!", status="done")
    session_view.add_operation(done_log)

    assert session_view.startup_logs is not None
    assert (
        session_view.startup_logs.content
        == "Starting process... more content done!"
    )
    assert session_view.startup_logs.status == "done"


def test_session_view_startup_logs_operations_exclude_done(
    session_view: SessionView,
) -> None:
    # Add startup log in progress
    start_log = StartupLogs(content="Starting...", status="start")
    session_view.add_operation(start_log)

    # Should include in operations while in progress
    operations = session_view.operations
    startup_ops = [op for op in operations if isinstance(op, StartupLogs)]
    assert len(startup_ops) == 1
    assert startup_ops[0].status == "start"

    # Mark as done
    done_log = StartupLogs(content=" complete", status="done")
    session_view.add_operation(done_log)

    # Should not include done startup logs in operations
    operations = session_view.operations
    startup_ops = [op for op in operations if isinstance(op, StartupLogs)]
    assert len(startup_ops) == 0


def test_session_view_startup_logs_standalone_done(
    session_view: SessionView,
) -> None:
    # Add a standalone "done" log without prior start/append
    done_log = StartupLogs(content="Process complete", status="done")
    session_view.add_operation(done_log)

    assert session_view.startup_logs is not None
    assert session_view.startup_logs.content == "Process complete"
    assert session_view.startup_logs.status == "done"


def test_session_view_package_logs_initialization(
    session_view: SessionView,
) -> None:
    """Test that SessionView initializes package_logs correctly."""
    assert hasattr(session_view, "package_logs")
    assert isinstance(session_view.package_logs, dict)
    assert len(session_view.package_logs) == 0


def test_session_view_package_logs_start(session_view: SessionView) -> None:
    """Test SessionView handles package logs start status."""

    alert = InstallingPackageAlert(
        packages={"numpy": "installing"},
        logs={"numpy": "Installing numpy...\n"},
        log_status="start",
    )

    session_view.add_operation(alert)

    assert "numpy" in session_view.package_logs
    assert session_view.package_logs["numpy"] == "Installing numpy...\n"


def test_session_view_package_logs_append(session_view: SessionView) -> None:
    """Test SessionView handles package logs append status."""

    # Start with initial log
    start_alert = InstallingPackageAlert(
        packages={"pandas": "installing"},
        logs={"pandas": "Starting installation...\n"},
        log_status="start",
    )
    session_view.add_operation(start_alert)

    # Append more logs
    append_alert = InstallingPackageAlert(
        packages={"pandas": "installing"},
        logs={"pandas": "Downloading dependencies...\n"},
        log_status="append",
    )
    session_view.add_operation(append_alert)

    expected_content = (
        "Starting installation...\nDownloading dependencies...\n"
    )
    assert session_view.package_logs["pandas"] == expected_content


def test_session_view_package_logs_done(session_view: SessionView) -> None:
    """Test SessionView handles package logs done status."""

    # Start installation
    start_alert = InstallingPackageAlert(
        packages={"scipy": "installing"},
        logs={"scipy": "Installing scipy...\n"},
        log_status="start",
    )
    session_view.add_operation(start_alert)

    # Add progress log
    append_alert = InstallingPackageAlert(
        packages={"scipy": "installing"},
        logs={"scipy": "Building wheels...\n"},
        log_status="append",
    )
    session_view.add_operation(append_alert)

    # Finish installation
    done_alert = InstallingPackageAlert(
        packages={"scipy": "installed"},
        logs={"scipy": "Successfully installed scipy!\n"},
        log_status="done",
    )
    session_view.add_operation(done_alert)

    expected_content = (
        "Installing scipy...\n"
        "Building wheels...\n"
        "Successfully installed scipy!\n"
    )
    assert session_view.package_logs["scipy"] == expected_content


def test_session_view_package_logs_multiple_packages(
    session_view: SessionView,
) -> None:
    """Test SessionView handles logs for multiple packages simultaneously."""

    # Start installing multiple packages
    multi_alert = InstallingPackageAlert(
        packages={"numpy": "installing", "pandas": "installing"},
        logs={
            "numpy": "Starting numpy install...\n",
            "pandas": "Starting pandas install...\n",
        },
        log_status="start",
    )
    session_view.add_operation(multi_alert)

    # Add logs for numpy only
    numpy_alert = InstallingPackageAlert(
        packages={"numpy": "installing", "pandas": "installing"},
        logs={"numpy": "Numpy progress...\n"},
        log_status="append",
    )
    session_view.add_operation(numpy_alert)

    # Add logs for pandas only
    pandas_alert = InstallingPackageAlert(
        packages={"numpy": "installing", "pandas": "installing"},
        logs={"pandas": "Pandas progress...\n"},
        log_status="append",
    )
    session_view.add_operation(pandas_alert)

    assert len(session_view.package_logs) == 2
    assert "numpy" in session_view.package_logs
    assert "pandas" in session_view.package_logs

    assert session_view.package_logs["numpy"] == (
        "Starting numpy install...\nNumpy progress...\n"
    )
    assert session_view.package_logs["pandas"] == (
        "Starting pandas install...\nPandas progress...\n"
    )


def test_session_view_package_logs_without_logs(
    session_view: SessionView,
) -> None:
    """Test SessionView handles InstallingPackageAlert without logs (backward compatibility)."""

    # Old-style alert without logs
    alert = InstallingPackageAlert(packages={"requests": "installing"})
    session_view.add_operation(alert)

    # Should not add any package logs
    assert len(session_view.package_logs) == 0


def test_session_view_package_logs_partial_logs(
    session_view: SessionView,
) -> None:
    """Test SessionView handles alerts with logs but no log_status."""

    # Alert with logs but no log_status
    alert = InstallingPackageAlert(
        packages={"matplotlib": "installing"},
        logs={"matplotlib": "Some log content...\n"},
        # log_status is None
    )
    session_view.add_operation(alert)

    # Should not add any package logs since log_status is missing
    assert len(session_view.package_logs) == 0


def test_session_view_package_logs_start_without_existing(
    session_view: SessionView,
) -> None:
    """Test package logs start status on package that doesn't exist yet."""

    alert = InstallingPackageAlert(
        packages={"new_package": "installing"},
        logs={"new_package": "Starting fresh install...\n"},
        log_status="start",
    )
    session_view.add_operation(alert)

    assert (
        session_view.package_logs["new_package"]
        == "Starting fresh install...\n"
    )


def test_session_view_package_logs_append_without_existing(
    session_view: SessionView,
) -> None:
    """Test package logs append status on package that doesn't exist yet."""

    # Append to non-existing package should start with empty string
    alert = InstallingPackageAlert(
        packages={"orphan_package": "installing"},
        logs={"orphan_package": "Appending to nothing...\n"},
        log_status="append",
    )
    session_view.add_operation(alert)

    assert (
        session_view.package_logs["orphan_package"]
        == "Appending to nothing...\n"
    )


def test_session_view_package_logs_empty_content(
    session_view: SessionView,
) -> None:
    """Test SessionView handles empty log content."""

    alert = InstallingPackageAlert(
        packages={"empty_logs": "installing"},
        logs={"empty_logs": ""},
        log_status="start",
    )
    session_view.add_operation(alert)

    assert "empty_logs" in session_view.package_logs
    assert session_view.package_logs["empty_logs"] == ""
